package cn.iotab.iotkit.transport.api.access;

import cn.iotab.iotkit.common.concurrent.NamedThreadFactory;
import cn.iotab.iotkit.queue.queue.QueueConsumer;
import cn.iotab.iotkit.queue.queue.QueueProducer;
import cn.iotab.iotkit.queue.rpc.QueueResponseTemplate;
import cn.iotab.iotkit.transport.api.access.handler.adaptor.ToServerMsgHandlerAdaptor;
import cn.iotab.iotkit.transport.api.access.handler.adaptor.ToServerRpcHandlerAdaptor;
import cn.iotab.iotkit.transport.api.common.message.JsonQueueMsg;
import cn.iotab.iotkit.transport.api.common.message.RpcRequestMessage;
import cn.iotab.iotkit.transport.api.common.message.RpcResponseMessage;
import cn.iotab.iotkit.transport.api.common.message.ToServerMessage;
import cn.iotab.iotkit.transport.api.common.message.ToTransportMessage;
import cn.iotab.iotkit.transport.api.common.message.codec.QueueMessageFactory;
import cn.iotab.iotkit.transport.api.common.message.msg.rpc.AuthResponseMsg;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionInfo;
import cn.iotab.iotkit.transport.api.common.queue.SendResultCallback;
import cn.iotab.iotkit.transport.api.common.queue.TransportQueueCallback;
import cn.iotab.iotkit.transport.api.common.queue.TransportQueueFactory;
import com.alibaba.fastjson.JSON;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class DefaultTransportAccessService implements TransportAccessService {

    private final TransportAccessConfig config;

    private final TransportQueueFactory queueProvider;

    //RPC template
    private QueueResponseTemplate<JsonQueueMsg<RpcRequestMessage>, JsonQueueMsg<RpcResponseMessage>> serverRpcResponseTemplate;
    // producer
    private QueueProducer<JsonQueueMsg<ToTransportMessage>> serverMsgProducer;
    // consumer
    private QueueConsumer<JsonQueueMsg<ToServerMessage>> serverNotificationsConsumer;

    private ExecutorService mainConsumerExecutor;

    //执行callback线程池
    protected ExecutorService callbackExecutor;

    private volatile boolean stopped = false;

    private ToServerRpcHandlerAdaptor toServerRpcHandlerAdaptor;
    private ToServerMsgHandlerAdaptor toServerMsgHandlerAdaptor;

    public DefaultTransportAccessService(TransportQueueFactory queueProvider,
                                         TransportAccessConfig config) {
        this.queueProvider = queueProvider;
        this.config = config;
    }

    @PostConstruct
    public void init(ToServerRpcHandlerAdaptor toServerRpcHandlerAdaptor,
                     ToServerMsgHandlerAdaptor toServerMsgHandlerAdaptor) {
        this.toServerRpcHandlerAdaptor = toServerRpcHandlerAdaptor;
        this.toServerMsgHandlerAdaptor = toServerMsgHandlerAdaptor;

        this.callbackExecutor = Executors.newWorkStealingPool(20);

        serverRpcResponseTemplate = queueProvider.createTransportApiResponseTemplate(config.getRpcChannelId());
        serverMsgProducer = queueProvider.createToTransportMsgProducer(config.getDownStreamTopic());
        serverNotificationsConsumer = queueProvider.createServerNotificationsConsumer(config.getUpStreamTopic(), config.getConsumerGroup());

        serverRpcResponseTemplate.init(this);
        serverMsgProducer.init();
        serverNotificationsConsumer.subscribe();

        mainConsumerExecutor = Executors.newSingleThreadExecutor(NamedThreadFactory.forNamePrefix("transport-consumer"));
        mainConsumerExecutor.execute(() -> {
            while (!stopped) {
                try {
                    List<JsonQueueMsg<ToServerMessage>> records = serverNotificationsConsumer.poll(config.getNotificationsPollDuration());
                    if (records.size() == 0) {
                        continue;
                    }
                    records.forEach(record -> {
                        try {
                            processToServerMsg(record.getValue());
                        } catch (Throwable e) {
                            log.warn("Failed to process the notification.", e);
                        }
                    });
                    serverNotificationsConsumer.commit();
                } catch (Exception e) {
                    if (!stopped) {
                        log.warn("Failed to obtain messages from queue.", e);
                        try {
                            Thread.sleep(config.getNotificationsPollDuration());
                        } catch (InterruptedException e2) {
                            log.trace("Failed to wait until the server has capacity to handle new requests", e2);
                        }
                    }
                }
            }
        });
    }

    @Override
    public void processToServerMsg(ToServerMessage toServerMessage) {
        callbackExecutor.submit(() -> {
            try {
                SessionInfo sessionInfo = toServerMessage.getSessionInfo();
                if (sessionInfo == null || sessionInfo.getSessionId() == null){
                    log.error("session is missing, ignore {}", toServerMessage);
                    return;
                }
                //process
                if (toServerMessage.getSessionEvent() != null) {
                    toServerMsgHandlerAdaptor.onDeviceSessionStateChange(sessionInfo, toServerMessage.getSessionEvent());
                    return;
                }
                if (toServerMessage.getSessionReportInfo() != null) {
                    toServerMsgHandlerAdaptor.onSessionReport(sessionInfo, toServerMessage.getSessionReportInfo());
                    return;
                }
                if (toServerMessage.getFromDeviceRequest() != null) {
                    toServerMsgHandlerAdaptor.onFromDeviceRequest(sessionInfo, toServerMessage.getFromDeviceRequest());
                    return;
                }
                if (toServerMessage.getFromDeviceResponse() != null) {
                    toServerMsgHandlerAdaptor.onFromDeviceResponse(toServerMessage.getFromDeviceResponse());
                    return;
                }
            }catch (Throwable t){
                log.error("process toServerMessage exception", t);
            }
            log.error("unknown ToServerMessage:{}", toServerMessage);
        });
    }

    @Override
    public void sendToTransportMsg(ToTransportMessage toTransportMessage, SendResultCallback<Void> callback) {
        String key = null;
        if (log.isTraceEnabled()) {
            log.trace("[{}] Pushing message to core: {}", key, toTransportMessage);
        }
        TransportQueueCallback queueCallback = callback != null ? new TransportQueueCallback(callback, callbackExecutor) : null;
        //
        JsonQueueMsg<ToTransportMessage> msg = new JsonQueueMsg<>(key, toTransportMessage);
        serverMsgProducer.send(msg, queueCallback);
    }


    @Override
    public ListenableFuture<JsonQueueMsg<RpcResponseMessage>> handle(JsonQueueMsg<RpcRequestMessage> msg) {
        RpcRequestMessage request = msg.getValue();
        String key = null;
        if (request.getAuthBasic() != null) {
            ListenableFuture<AuthResponseMsg> responseMsg = toServerRpcHandlerAdaptor.onBasicAuth(request.getAuthBasic());
            return Futures.transform(responseMsg,
                    value -> QueueMessageFactory.createQueueMessage(key, RpcResponseMessage.builder().authResponse(value).build()),
                    MoreExecutors.directExecutor());
        }
        if (request.getAuthX509() != null) {
            ListenableFuture<AuthResponseMsg> responseMsg = toServerRpcHandlerAdaptor.onX509Auth(request.getAuthX509());
            return Futures.transform(responseMsg,
                    value -> QueueMessageFactory.createQueueMessage(key, RpcResponseMessage.builder().authResponse(value).build()),
                    MoreExecutors.directExecutor());
        }
        //
        log.warn("unknown RpcRequestMessage:" + JSON.toJSONString(request));
        return null;
    }

    @PreDestroy
    public void destroy() {
        stopped = true;

        if (serverNotificationsConsumer != null) {
            serverNotificationsConsumer.unsubscribe();
        }
        if (serverMsgProducer != null) {
            serverMsgProducer.stop();
        }
        if (mainConsumerExecutor != null) {
            mainConsumerExecutor.shutdownNow();
        }
        if (serverRpcResponseTemplate != null) {
            serverRpcResponseTemplate.stop();
        }
    }
}
